package com.asiainfo.dacp.datastash.worker.executor.adapter.datax.translate;

import java.util.Map;

import com.asiainfo.dacp.datastash.worker.executor.adapter.datax.model.domain.Job;
import com.asiainfo.dacp.datastash.worker.executor.adapter.datax.model.domain.JobId;
import com.asiainfo.dacp.datastash.worker.executor.adapter.datax.model.plugin.meta.DpMeta;
import com.asiainfo.dacp.datastash.worker.executor.adapter.datax.model.plugin.meta.FtpMeta;
import com.asiainfo.dacp.datastash.worker.executor.adapter.datax.model.plugin.meta.HdfsMeta;
import com.asiainfo.dacp.datastash.worker.executor.adapter.datax.utils.CommonErrorCode;
import com.asiainfo.dacp.datastash.worker.executor.adapter.datax.utils.Configuration;
import com.asiainfo.dacp.metamodel.domain.ds.model.MetaDataSource;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;

public abstract class JobTranslate {
	public abstract Job translate(JobId id) ;
	
	protected FtpMeta buildFtpMeta(MetaDataSource ftpDs){
		FtpMeta ftp = new FtpMeta();
		String dsConfStr = ftpDs.getDsConf();
		Configuration dsConf = Configuration.from(dsConfStr);
		ftp.setProtocol(dsConf.getString("protocol", "sftp"));
		ftp.setHost(dsConf.getNecessaryValue("ip", CommonErrorCode.CONFIG_ERROR));
		ftp.setPort(dsConf.getInt("port", 22));
		String dsAcct = ftpDs.getDsAcct();
		String dsAuth = ftpDs.getDsAuth();
		Preconditions.checkNotNull(dsAcct, "FTP 用户名为空 .");
		Preconditions.checkNotNull(dsAuth, "FTP 密码为空 .");
		ftp.setUsername(dsAcct);
		ftp.setPassword(dsAuth);
		return ftp;	
	}
	protected DpMeta buildDpMeta(Configuration unitConf){
		DpMeta dp=new DpMeta();
		dp.setProcName(unitConf.getString("procName"));
		dp.setProcParam(unitConf.getString("procParam"));
		return dp;
	}
	HdfsMeta buildHdfsMeta(MetaDataSource hdfsDs, Configuration oriConf){
		HdfsMeta hdfs = new HdfsMeta();
		String dsConfStr = hdfsDs.getDsConf();
		Configuration conf = Configuration.from(dsConfStr);
		hdfs.setDefaultFS(conf.getString("nameService"));
		hdfs.setHaveKerberos(conf.getString("isKerberos", "N").equalsIgnoreCase("Y"));
		hdfs.setKerberosKeytabFilePath(conf.getString("kerberosKeytabFilePath"));
		hdfs.setKerberosPrincipal(conf.getString("kerberosPrincipal"));
		
		if(!Strings.isNullOrEmpty(hdfsDs.getDsAcct() )){
			System.setProperty("HADOOP_USER_NAME", hdfsDs.getDsAcct());
		}
		
		hdfs.setFileType(oriConf.getString("fileType", "text").toUpperCase());
		
		String hcfg = oriConf.getString("hadoopConfig");
		if(!Strings.isNullOrEmpty(hcfg)){
			hdfs.setHadoopConfig(conf.getMap("hadoopConfig"));
		} 
		
		return hdfs;
	}
	
}



